package com.lcz.manage.util.sharding;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Collections;
import java.util.List;

import org.springframework.util.CollectionUtils;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.lcz.manage.util.exception.CCException;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lucz@yunrong.cn
 * @version V3.0
 * @date 2021/9/24 14:31
 */
@Slf4j
public class ShardingUtils {

    /**
     * 大文件分片信息记录，步骤一
     * @param localFilePath
     * @param fixNum
     * @return
     */
    public static List<ShardingBlock> nioFileSharding(String localFilePath, int fixNum){
        //读取文件
        File file = new File(localFilePath);
        if (!file.exists()) {
            log.info("本地路径 {} 下文件不存在", localFilePath);
            throw new CCException("文件不存在");
        }
        List<ShardingBlock> shardingBlockList = Lists.newArrayList();
        try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
            long rafLength = raf.length();
            if(rafLength == 0L){
                log.info("文件内容为空");
                return Collections.emptyList();
            }
            long t1 = System.currentTimeMillis();
            // 获取RandomAccessFile对象文件指针的位置，初始位置为0
            Long pointer = 0L;
            // 移动文件指针位置
            raf.seek(pointer);
            FileChannel fileChannel = raf.getChannel();
            ByteBuffer readByteBuffer = ByteBuffer.allocate(10000);

            // 行号
            int lineNumber = 0;
            // 最后一次分片标记行号
            int lastMarkLineNum = 0;
            //最后一次分片标记坐标
            long lastMarkPointer = 0;
            // 分片流水
            int batchNum = 0;

            final int LF = 10;
            final int CR = 13;
            byte[] lineByte = new byte[0];
            byte[] temp = new byte[0];
            while ((fileChannel.read(readByteBuffer) != -1)) {
                int readSize = readByteBuffer.position();
                byte[] bs = new byte[readSize];
                readByteBuffer.rewind();
                readByteBuffer.get(bs);
                readByteBuffer.clear();

                int startNum = 0;
                boolean hasLF = false;
                for (int i = 0; i < readSize; i++) {
                    if (bs[i] == LF
                        // 修复只有一行内容且没有回车读不到数据的场景。如果影响其他可以去掉另行优化
                        || (rafLength - 1L) == i) {
                        hasLF = true;
                        int tempNum = temp.length;
                        int lineNum = i - startNum;
                        lineByte = new byte[tempNum + lineNum];
                        System.arraycopy(temp, 0, lineByte, 0, tempNum);
                        temp = new byte[0];
                        System.arraycopy(bs, startNum, lineByte, tempNum, lineNum);
                        if (lineNumber % fixNum == 0) {

                            lastMarkLineNum = lineNumber;
                            lastMarkPointer = pointer;
                            FileLineData fileLineData = new FileLineData();
                            fileLineData.setLineNum(lastMarkLineNum + 1);
                            fileLineData.setPointer(lastMarkPointer);
                            fileLineData.setNextLineNum(fixNum);
                            fileLineData.setFileName(file.getName());

                            // 流水号批次号叠加
                            batchNum++;
                            ShardingBlock shardingBlock = ShardingBlock.builder().build();
                            shardingBlock.setBlockIndex(batchNum);
                            shardingBlock.setStartNumber(lastMarkLineNum + 1);
                            shardingBlock.setEndNumber(lastMarkLineNum + fileLineData.getNextLineNum());
                            shardingBlock.setLength(fileLineData.getNextLineNum());
                            shardingBlock.setPointer(pointer);
                            // 将原始查询条件 赋值到额外参数中
                            shardingBlock.setExtParam(JSON.toJSONString(fileLineData));
                            shardingBlockList.add(shardingBlock);
                        }
                        // 游标前移
                        pointer = raf.getFilePointer() - readSize + i + 1;
                        // 当前行
                        ++lineNumber;

                        if (i + 1 < readSize && bs[i + 1] == CR) {
                            startNum = i + 2;
                            pointer++;
                        } else {
                            startNum = i + 1;
                        }
                    }
                }
                if (hasLF) {
                    temp = new byte[bs.length - startNum];
                    System.arraycopy(bs, startNum, temp, 0, temp.length);
                } else {
                    byte[] toTemp = new byte[temp.length + bs.length];
                    System.arraycopy(temp, 0, toTemp, 0, temp.length);
                    System.arraycopy(bs, 0, toTemp, temp.length, bs.length);
                    temp = toTemp;
                }
            }
            // 判断最后一个分片，更新最后一片的数据的行数
            if (!CollectionUtils.isEmpty(shardingBlockList)) {
                int nextLineNum = lineNumber - lastMarkLineNum + 1;
                FileLineData fileLineData = JSON.parseObject(shardingBlockList.get(shardingBlockList.size() - 1)
                    .getExtParam(), FileLineData.class);
                fileLineData.setPointer(lastMarkPointer);
                fileLineData.setNextLineNum(nextLineNum);

                ShardingBlock lastBlock = shardingBlockList.get(shardingBlockList.size() - 1);
                lastBlock.setExtParam(JSON.toJSONString(fileLineData));
                lastBlock.setLength(nextLineNum);
                lastBlock.setEndNumber(lastBlock.getStartNumber() + nextLineNum - 1);
            }
            log.info("文件分片数：{}", shardingBlockList.size());
            log.info("文件分片总耗时：{}ms", (System.currentTimeMillis() - t1));
        } catch (Exception ex) {
            throw new CCException("读取文件信息失败");
        }
        return shardingBlockList;
    }

    /**
     * 大文件分片读取，步骤二
     *
     * @param pointer
     * @param nextLineNum
     * @param localFilePath
     * @return
     */
    public static List<String> nioShardingRead(long pointer, int nextLineNum, String localFilePath) {
        List<String> shardingResult = Lists.newArrayList();
        //原始查询条件

        //读取文件
        File file = new File(localFilePath);
        if (file == null || !file.exists()) {
            log.info("本地路径 {} 下文件不存在", localFilePath);
            throw new CCException("文件不存在");
        }
        log.info("开始读取文件 {} ,游标位置为 {}, 往后读取 {} 行数据处理", localFilePath, pointer, nextLineNum);
        //遍历文件内容 赋值
        List<String> creditSliceBatchLogDoList = Lists.newArrayList();
        try (RandomAccessFile raf = new RandomAccessFile(file, "r");) {
            // 获取RandomAccessFile对象文件指针的位置，初始位置是0
            // 移动文件指针位置
            raf.seek(pointer);

            //行内容
            String line;
            // 循环读取
            while ((line = raf.readLine()) != null) {
                //文本转为utf-8
                creditSliceBatchLogDoList.add(new String(line.getBytes("ISO-8859-1"), "UTF-8"));
                //游标前移
                pointer = raf.getFilePointer();
                //行号叠加
                if (--nextLineNum <= 0) {
                    break;
                }
            }

        } catch (IOException e) {
            log.info("分片信息错误，错误信息：{}", e.getMessage());
            throw new CCException("分片信息错误");
        }
        shardingResult.addAll(creditSliceBatchLogDoList);
        return shardingResult;
    }


}
